package indi.zhifa.recipe.bailan.framework.queue.util;

import cn.hutool.core.convert.Convert;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.support.config.FastJsonConfig;
import indi.zhifa.recipe.bailan.framework.queue.entity.EQueueValueType;
import indi.zhifa.recipe.bailan.framework.redis.config.RedisConfig;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

@Component
public class ZfRedisMqUtil {

    public static final String TYPE_KEY = "type";
    public static final String MSG_ID_KEY = "id";
    public static final String DATA_KEY = "data";

    private final StreamOperations<String,String,Object> mStreamOperations;

    public ZfRedisMqUtil(RedisConnectionFactory factory){
        RedisTemplate<String, Object> redisTemplate = genTemplate(factory);
        mStreamOperations = redisTemplate.opsForStream();
    }

    public RedisTemplate<String,Object> genTemplate(RedisConnectionFactory factory){
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
        redisTemplate.setConnectionFactory(factory);
        redisTemplate.setKeySerializer(RedisSerializer.string());
        redisTemplate.setValueSerializer(RedisSerializer.string());
        redisTemplate.setHashKeySerializer(RedisSerializer.string());
        redisTemplate.setHashValueSerializer(RedisSerializer.byteArray());
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }

    public void sendMessage(String pStream , Integer pMsgId, EQueueValueType pType, Object pData){
        Map<String,Object> data = new HashMap<String,Object>();
        data.put(MSG_ID_KEY,Convert.intToBytes(pMsgId));
        data.put(TYPE_KEY,Convert.intToBytes(pType.getCode()));
        FastJsonConfig fastJsonConfig = RedisConfig.getRedisFastJson();
        byte[] byteData;
        switch (pType){
            case DEFAULT:
            case STRING:
                byteData = pData.toString().getBytes(StandardCharsets.UTF_8);
                break;
            case JSON:
                byteData = JSON.toJSONBytes(pData,fastJsonConfig.getWriterFeatures());
                break;
            case INT:
                byteData = Convert.intToBytes((int)pData);
                break;
            case LONG:
                byteData = Convert.longToBytes((long)pData);
                break;
            default:
                byteData = null;
                break;
        }
        data.put(DATA_KEY,byteData);
        mStreamOperations.add(pStream,data);
    }

}
